package com.pw.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.pw.gmall.realtime.app.BaseApp;
import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.utils.AtguiguUtils;
import com.pw.gmall.realtime.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Comparator;
import java.util.List;

/**
 * @Author: linux_future
 * @since: 2022/3/15
 **/
public class DwdLogApp extends BaseApp {

    public static void main(String[] args) {
        new DwdLogApp().init(2001, 1, "DwdLogApp", "DwdLogApp", Constant.TOPIC_ODS_LOG);
    }

    @Override
    protected void handler(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        System.out.println("handler.........................");
        //处理数据
        //1.校验是否是新用户
        SingleOutputStreamOperator<JSONObject> streamCheckUser = validateNewUser(stream);
        //streamCheckUser.print();

        // 2. 流的拆分: 利用侧输出流
        Tuple3<DataStream<JSONObject>, SingleOutputStreamOperator<JSONObject>, DataStream<JSONObject>> streamSplit = splitStream(streamCheckUser);
        streamSplit.f0.print("page:");
        streamSplit.f1.print("start:");
        streamSplit.f2.print("display:");

        // 3. 分别写入到不同的Topic中
        sinkKafka(streamSplit);

    }

    private void sinkKafka(Tuple3<DataStream<JSONObject>, SingleOutputStreamOperator<JSONObject>, DataStream<JSONObject>> stream) {
        //写入kafka
        stream.f0.map(JSONAware::toJSONString).addSink(KafkaUtils.getKafkaSink(Constant.TOPIC_DWD_PAGE_LOG));
        stream.f1.map(JSONAware::toJSONString).addSink(KafkaUtils.getKafkaSink(Constant.TOPIC_DWD_START_LOG));
        stream.f2.map(JSONAware::toJSONString).addSink(KafkaUtils.getKafkaSink(Constant.TOPIC_DWD_DISPLAY_LOG));

    }

    private Tuple3<DataStream<JSONObject>, SingleOutputStreamOperator<JSONObject>, DataStream<JSONObject>> splitStream(SingleOutputStreamOperator<JSONObject> stream) {
        //页面日志、启动日志和曝光日志,使用测输出流分流
        OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("page") {};
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("display") {};
        /*
         启动日志放主流
         页面和曝光分表放入到一个测输出流
         */

        SingleOutputStreamOperator<JSONObject> startStream = stream.process(
                new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        // 1. 判断是否为启动日志
                        if (value.containsKey("start")) {
                            out.collect(value);
                        } else {
                            // 2. 判断是否页面
                            if (value.containsKey("page")) {
                                ctx.output(pageTag, value);
                            }

                            // 3. 判断是否为曝光日志
                            if (value.containsKey("displays")) {
                                // 因为曝光日志中的曝光数据是数组, 最好展开, 分别放入流中
                                JSONArray array = value.getJSONArray("displays");
                                for (int i = 0; i < array.size(); i++) {
                                    JSONObject display = array.getJSONObject(i);

                                    // 给曝光数据增添一些其他字段
                                    display.putAll(value.getJSONObject("common"));
                                    display.putAll(value.getJSONObject("page"));
                                    display.put("ts", value.getLong("ts"));

                                    ctx.output(displayTag, display);
                                }
                            }

                        }

                    }
                });


        DataStream<JSONObject> streamPage = startStream.getSideOutput(pageTag);
        DataStream<JSONObject> streamDisplay = startStream.getSideOutput(displayTag);

        return Tuple3.of(streamPage, startStream, streamDisplay);

    }

    private SingleOutputStreamOperator<JSONObject> validateNewUser(DataStreamSource<String> stream) {
        //
        WatermarkStrategy<JSONObject> strategy = WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((obj, ts) -> obj.getLong("ts"));

        SingleOutputStreamOperator<JSONObject> result = stream.map(JSON::parseObject).assignTimestampsAndWatermarks(strategy)
                .keyBy(obj -> obj.getJSONObject("common").getString("mid"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {

                    private ValueState<Long> firstVisitState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstVisitState = getRuntimeContext().getState(new ValueStateDescriptor<>("firstVisitState", Long.class));
                    }

                    @Override
                    public void process(String key, Context ctx, Iterable<JSONObject> elements, Collector<JSONObject> out) throws Exception {
                        Long firstVisitValue = this.firstVisitState.value();

                        if (firstVisitValue == null) {
//把窗口中的数据排序
                            List<JSONObject> list = AtguiguUtils.toList(elements);
                            list.sort(Comparator.comparing(o -> o.getLong("ts")));
                            for (int i = 0; i < list.size(); i++) {
                                //第一个状态修改为1，并保存在state中
                                if (i == 0) {
                                    //更新状态为1
                                    list.get(i).getJSONObject("common").put("is_new", "1");
                                    //保存第一次访问时间
                                    firstVisitState.update(list.get(i).getLong("ts"));
                                } else {
                                    list.get(i).getJSONObject("common").put("is_new", "0");
                                }
                                out.collect(list.get(i));
                            }
                        } else {
                            //不为空，说明设备已来过，设备状态全更新为0
                            elements.forEach(ele -> {
                                ele.getJSONObject("common").put("is_new", "0");
                                out.collect(ele);
                            });

                        }
                    }
                });
        return result;
    }
}
